/*
 * Copyright © 2020 https://www.lrting.top/ All rights reserved.
 */


package com.zh.ch.bigdata.flink;

import com.zh.ch.bigdata.flink.sqlsubmit.EnumRunningMode;
import com.zh.ch.bigdata.flink.sqlsubmit.SqlParser;
import org.apache.commons.cli.CommandLine;
import org.apache.commons.cli.CommandLineParser;
import org.apache.commons.cli.DefaultParser;
import org.apache.commons.cli.Options;
import org.apache.commons.cli.ParseException;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.TableEnvironment;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Locale;
import java.util.Map;

/**
 * @author: bigDataToAi
 * @date: 2022/7/8
 * @description:
 * @modifiedBy:
 * @version: 1.0
 */
public class SqlSubmit {

    private static final Logger LOG = LoggerFactory.getLogger(SqlSubmit.class);

    public static CommandLine parseArgs(String[] args) throws ParseException {
        CommandLineParser commandLineParser = new DefaultParser();
        Options options = new Options();
        options.addOption("f","sqlFilePath", true, "sql文件路径");
        options.addOption("m", "mode", true, "flink运行模式，batch/streaming");
        return commandLineParser.parse(options, args);
    }


    public static void main(String[] args) throws Exception {

        CommandLine commandLine = parseArgs(args);
        String sqlFilePath;
        if (commandLine.hasOption("f")) {
            sqlFilePath = commandLine.getOptionValue("f");
        }
        else {
            throw new Exception("请指定sql文件路径， -f xxx.sql");
        }
        String runningMode = commandLine.getOptionValue("m", EnumRunningMode.STREAMING.name().toLowerCase(Locale.ROOT));

        EnvironmentSettings settings;
        if (EnumRunningMode.BATCH.name().equals(runningMode.toUpperCase())) {
            settings = EnvironmentSettings
                    .newInstance()
                    .inBatchMode()
                    .build();
        }
        else if (EnumRunningMode.STREAMING.name().equals(runningMode.toUpperCase())) {
            settings = EnvironmentSettings
                    .newInstance()
                    .inStreamingMode()
                    .build();
        }
        else {
            throw new Exception("不支持指定的flink运行模式：" + runningMode);
        }

        TableEnvironment tableEnvironment = TableEnvironment.create(settings);

        for (Map<String, String> sql : SqlParser.parseSqlFile(sqlFilePath)) {
            if (sql != null && sql.size() != 0) {
                LOG.info(sql.values().iterator().next());
                tableEnvironment.executeSql(sql.values().iterator().next());
            }
        }
    }

}

